/*
 * Copyright 2020-2025 the original author or authors.
 * You cannot use this file unless authorized by the author.
 */
package org.ipig.commons.conf.kafka.consumer;

import lombok.Data;
import lombok.ToString;
import org.ipig.commons.helper.StringHelper;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * kafka消费者配置
 *
 * @author <a href="mailto:comchnts@163.com">chinats</a>
 * @version $Id: KafkaConsumerConf.java 3 2020-01-21 09:36:33Z ts <comchnts@163.com> $
 * @since 1.0
 */
@Data
@ToString
public class KafkaConsumerConf implements Serializable {
    //如果有多个topic请创建多个KafkaConsumerConf实例
    private String topicName = "";
    private String bootstrapServers = "";
    private String groupId = "";
    private String autoCommitEnable = "true";
    private String keyDeserializer = "org.apache.kafka.common.serialization.StringDeserializer";
    private String valueDeserializer = "org.apache.kafka.common.serialization.StringDeserializer";
    private String maxPollRecords = "300";/*控制一个poll()调用返回的记录数，这个可以用来控制应用在拉取循环中的处理数据量。*/
    private String autoCommitMs = "1000";//consumer向ZK或服务器提交offset的频率，单位是毫秒
    private String sessionTimeoutMs = "60000";//用于检查worker故障的超时时间。worker定时发送心跳以表明自己是活着的。如果broker在会话超时时间到期之前没有接收到心跳，那么broker将从分组中移除该worker，并启动重新平衡。注意，该值必须在group.min.session.timeout.ms【默认6000】和group.max.session.timeout.ms【默认300000】范围内。
    private String maxPollIntervalMs = "300000";//最大的poll数据间隔，如果超过这个间隔没有发起poll请求，但heartbeat仍旧在发，就认为该consumer处于活锁(livelock)状态。就会将该consumer退出consumer group。所以为了不使Consumer 自己被退出，Consumer 应该不停的发起poll(timeout)操作。而这个动作 KafkaConsumer Client是不会帮我们做的，这就需要自己在程序中不停的调用poll方法了。
    private String requestTimeoutMs = "30000";  //从发送请求到收到ACK确认等待的最长时间（超时时间）,如果在超时时间结束之前没有收到响应，则客户端将在必要时重新发送请求，或者在重试次数用完时使请求失败。
    private String fetchMaxWaitMs = "500";  //如果没有足够的数据立即满足fetch.min.bytes给出的要求，服务器在回答获取请求之前将阻塞的最大时间【最大不能超过500ms，超过后统一赋值为500ms】
    private String heartbeatIntervalMs = "3000";  //使用kafka的群组管理设施时，消费者协调器心跳的预期时间。心跳用于确保消费者的会话保持活跃，并在新消费者加入或离开组时促进重新平衡。该值必须设置为低于session.timeout.ms，但通常应设置不高于该值的1/3。它可以调整得更低，以控制正常再平衡的预期时间。
    private String connectionsMaxIdleMs = "540000";  //Close idle connections after the number of milliseconds specified by this config.
    private String defaultApiTimeoutMs = "60000";  //kafka connect中加入此参数使得partitionsFor, listTopics, offsetsForTimes, beginningOffsets, endOffsets，close不用无限阻塞
    private String metadataMaxAgeMs = "300000";  //以微秒为单位的时间，强制更新metadata的时间间隔。即使我们没有看到任何partition leadership改变。
    private String metricsSampleWindowMs = "30000";  //metrics系统维护可配置的样本数量，在一个可修正的window  size

    /**
     * earliest: 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
     * latest: 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据
     * none：topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常
     */
    private String autoOffsetReset = "earliest";
    //读取数据时使用的TCP接收缓冲区大小
    //其它参数,org.apache.kafka.clients.CommonClientConfigs
//    private Map<String, String> map = new HashMap<String, String>();
    // 其它属性
    private Map<String, Object> props = new HashMap<>();
    //KafkaOffsetsInfo
    private List<KafkaOffsetsInfo> offsetList = new ArrayList<>();

    // 兼容之前的API
    @Deprecated
    public Map<String, String> getMap() {
        Map<String, String> map = new HashMap<String, String>();
        for (Map.Entry<String, Object> e : props.entrySet()) {
            map.put(e.getKey(), StringHelper.valueOf(e.getValue()));
        }
        return map;
    }

    // 兼容之前的API
    @Deprecated
    public void setMap(Map<String, String> inMap) {
        props.putAll(inMap);
    }
}
